package com.imooc.spark.streaming.dao

import java.sql.{Connection, PreparedStatement, ResultSet}

import com.imooc.spark.streaming.entity.{OrderSaleAmount, OrderSaleVolume}
import com.imooc.spark.streaming.utils.MysqlUtils

import scala.collection.mutable.ListBuffer

/**
  * @description
  * @author yuyon26@126.com
  * @date 2018/11/6 11:13
  */
object OrderSaleMySqlDao {

  /**
    * 批量插入销售额
    *
    * @param list
    */
  def insertOrderSaleVolumeBatch(list: ListBuffer[OrderSaleVolume]) = {
    var connection: Connection = null
    var pstmt: PreparedStatement = null
    try {
      connection = MysqlUtils.getConnetction()

      val sql = "insert into sale_volume_info(c_id,day,money) values (?,?,?) "
      pstmt = connection.prepareStatement(sql)

      //关闭自动提交
      //connection.setAutoCommit(false)
      for (data <- list) {
        pstmt.setString(1, data.companyId)
        pstmt.setString(2, data.day)
        pstmt.setFloat(3, data.money + getOrderSaleVolume(data, connection))
        pstmt.addBatch()
        deleteOrderSaleVolume(data, connection)
        pstmt.executeUpdate()
      }
      //执行批处理
      //pstmt.executeBatch()
      //connection.commit()
    } catch {
      case e: Exception => e.printStackTrace()
    } finally {
      MysqlUtils.release(connection, pstmt)
    }
  }

  /**
    * 批量插入销售量
    *
    * @param list
    */
  def insertOrderSaleAmountBatch(list: ListBuffer[OrderSaleAmount]) = {
    var connection: Connection = null
    var pstmt: PreparedStatement = null
    try {
      connection = MysqlUtils.getConnetction()

      val sql = "insert into sale_amount_info(c_id,day,count) values (?,?,?) "
      pstmt = connection.prepareStatement(sql)

      //关闭自动提交
      //connection.setAutoCommit(false)
      for (data <- list) {
        pstmt.setString(1, data.companyId)
        pstmt.setString(2, data.day)
        pstmt.setFloat(3, data.count + getOrderSaleAmount(data, connection))
        pstmt.addBatch()
        deleteOrderSaleAmount(data, connection)
        pstmt.executeUpdate()
      }
      //执行批处理
      //pstmt.executeBatch()
      //connection.commit()
    } catch {
      case e: Exception => e.printStackTrace()
    } finally {
      MysqlUtils.release(connection, pstmt)
    }
  }

  /**
    * 根据c_id&day获取money
    *
    * @param orderSaleVolume
    */
  def getOrderSaleVolume(orderSaleVolume: OrderSaleVolume, connection: Connection): Float = {
    val sql = "select * from sale_volume_info where c_id=? and day=?"
    val pstmt: PreparedStatement = connection.prepareStatement(sql)
    pstmt.setString(1, orderSaleVolume.companyId)
    pstmt.setString(2, orderSaleVolume.day)
    val rs = pstmt.executeQuery()
    if (rs.next()) {
      rs.getFloat("money")
    } else {
      0F
    }
  }

  /**
    * 根据c_id&day获取count
    *
    * @param orderSaleAmount
    */
  def getOrderSaleAmount(orderSaleAmount: OrderSaleAmount, connection: Connection): Long = {
    val sql = "select * from sale_amount_info where c_id=? and day=?"
    val pstmt: PreparedStatement = connection.prepareStatement(sql)
    pstmt.setString(1, orderSaleAmount.companyId)
    pstmt.setString(2, orderSaleAmount.day)
    val rs = pstmt.executeQuery()
    if (rs.next()) {
      rs.getLong("count")
    } else {
      0L
    }
  }

  /**
    * 根据c_id&day删除记录
    *
    * @param orderSaleVolume
    */
  def deleteOrderSaleVolume(orderSaleVolume: OrderSaleVolume, connection: Connection) = {
    val sql = "DELETE FROM sale_volume_info WHERE c_id=? and day=?"
    val pstmt: PreparedStatement = connection.prepareStatement(sql)
    pstmt.setString(1, orderSaleVolume.companyId)
    pstmt.setString(2, orderSaleVolume.day)
    pstmt.executeUpdate()
  }

  /**
    * 根据c_id&day删除记录
    *
    * @param orderSaleAmount
    */
  def deleteOrderSaleAmount(orderSaleAmount: OrderSaleAmount, connection: Connection) = {
    val sql = "DELETE FROM sale_amount_info WHERE c_id=? and day=?"
    val pstmt: PreparedStatement = connection.prepareStatement(sql)
    pstmt.setString(1, orderSaleAmount.companyId)
    pstmt.setString(2, orderSaleAmount.day)
    pstmt.executeUpdate()
  }


  def main(args: Array[String]): Unit = {
    val connection: Connection = MysqlUtils.getConnetction()
    val money = OrderSaleMySqlDao.getOrderSaleVolume(OrderSaleVolume("1001", "20181111", 0F), connection)
    val count = OrderSaleMySqlDao.getOrderSaleAmount(OrderSaleAmount("1001", "20181111", 0L), connection)
    //OrderSaleMySqlDao.deleteOrderSaleAmount(OrderSaleAmount("1001", "20181111", 0L), connection)
    println(money)
    println(count)
  }

}
